/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package me.flyray.bsin.server.ws;


import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PreDestroy;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import me.flyray.bsin.domain.request.AppChatDTO;
import me.flyray.bsin.security.contex.LoginInfoContextHelper;
import me.flyray.bsin.security.filter.WebsocketLoginInfoInterceptor;
import me.flyray.bsin.server.biz.AppAgentEngine;
import me.flyray.bsin.server.biz.WebSocketAiChatAsyncBiz;
import me.flyray.bsin.utils.BsinSnowflake;
import org.apache.shenyu.client.spring.websocket.annotation.ShenyuServerEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * WebSocket服务端点 - 处理AI聊天连接
 * chatType: 0 非流式 1 流式
 */
@Component
@ShenyuServerEndpoint(value = "/{toNo}/{chatType}", configurator = WebsocketLoginInfoInterceptor.class)
public class WsServerEndpoint {

    private static final Logger LOG = LoggerFactory.getLogger(WsServerEndpoint.class);
    
    // JSON解析器 - 静态实例，线程安全
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    
    // 消息限制配置
    private static final int MAX_MESSAGE_SIZE = 10 * 1024; // 10KB
    private static final long MESSAGE_INTERVAL_MS = 100; // 消息间隔限制
    private static final int MAX_QUEUE_SIZE = 100; // 消息队列最大容量，防止内存泄漏

    // 线程池配置 - 优化版本，支持动态扩缩容和拒绝策略
    private static final ThreadPoolExecutor WEB_SOCKET_EXECUTOR = new ThreadPoolExecutor(
            5,                    // 核心线程数
            20,                   // 最大线程数
            60L, TimeUnit.SECONDS, // 空闲线程存活时间
            new LinkedBlockingQueue<>(1000), // 有界队列，防止OOM
            new ThreadFactory() {
                private final AtomicInteger threadNumber = new AtomicInteger(1);
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r, "websocket-ai-chat-" + threadNumber.getAndIncrement());
                    t.setDaemon(true);
                    t.setUncaughtExceptionHandler((thread, ex) -> 
                        LOG.error("WebSocket线程异常 - Thread: {}", thread.getName(), ex));
                    return t;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：调用者线程执行
    );

    // 在线连接数统计
    private static final AtomicInteger onlineCount = new AtomicInteger(0);

    // 客户端连接集合 - 线程安全
    private static final CopyOnWriteArraySet<WsServerEndpoint> wsClientMap = new CopyOnWriteArraySet<>();

    // session池 - 用户ID -> Session映射
    private static final ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
    
    // 连接管理锁 - 防止竞态条件
    private static final Object connectionLock = new Object();

    // Spring应用上下文 - 解决依赖注入问题
    private static ApplicationContext applicationContext;

    // 实例属性
    private Session session;
    private String fromNo;  // 发起人用户ID
    private String toNo;    // 目标对象ID
    private String chatType; // 聊天类型
    private volatile long lastMessageTime = 0; // 最后消息时间，用于限流
    private volatile boolean closed = false; // 连接是否已关闭，防止关闭后继续入队消息

    /**
     * 设置Spring应用上下文
     */
    public static void setApplicationContext(ApplicationContext context) {
        applicationContext = context;
    }

    /**
     * WebSocket连接建立时触发
     */
    @OnOpen
    public void onOpen(@PathParam("toNo") String toNo, @PathParam("chatType") String chatType, Session session, EndpointConfig config) {
        this.session = session;
        this.toNo = toNo;
        this.chatType = chatType;

        try {
            // 从配置中获取用户信息
            UserInfo userInfo = extractUserInfo(config);
            if (userInfo == null) {
                closeWithReason(session, "用户信息获取失败");
                return;
            }

            // 检查用户认证状态
            if (!userInfo.isAuthenticated()) {
                closeWithReason(session, "用户未认证");
                return;
            }

            // 设置登录上下文
            setupLoginContext(userInfo);
            this.fromNo = userInfo.getBizRoleTypeNo();

            // 处理重复登录 - 关闭旧连接
            handleDuplicateLogin(userInfo.getBizRoleTypeNo());

            // 注册新连接
            registerConnection(userInfo.getBizRoleTypeNo(), session);

            LOG.info("新WebSocket连接建立 - SessionId: {}, FromNo: {}, ToNo: {}, 在线数: {}",
                    session.getId(), fromNo, toNo, getOnlineCount());

        } catch (Exception e) {
            LOG.error("WebSocket连接建立失败", e);
            closeWithReason(session, "连接建立失败");
        }
    }

    /**
     * WebSocket连接关闭时触发 - 改进清理逻辑
     */
    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        try {
            synchronized (connectionLock) {
                // 标记连接已关闭，防止新消息入队
                closed = true;
                
                // 清理消息队列，防止内存泄漏
                int queueSize = messageQueue.size();
                messageQueue.clear();
                if (queueSize > 0) {
                    LOG.info("清理消息队列 - FromNo: {}, 丢弃消息数: {}", fromNo, queueSize);
                }
                
                // 从连接池中移除
                wsClientMap.remove(this);
                
                // 从session池中移除 - 改进清理逻辑
                if (StringUtils.hasText(fromNo)) {
                    sessionPool.remove(fromNo);
                    LOG.debug("从sessionPool移除连接 - 用户: {}", fromNo);
                } else {
                    // 如果fromNo为空，遍历sessionPool找到对应的session并移除
                    sessionPool.entrySet().removeIf(entry -> {
                        boolean shouldRemove = entry.getValue() != null && entry.getValue().equals(session);
                        if (shouldRemove) {
                            LOG.debug("从sessionPool移除匿名连接 - SessionId: {}", session.getId());
                        }
                        return shouldRemove;
                    });
                }
                
                // 减少在线数统计
                onlineCount.decrementAndGet();
                
                LOG.info("WebSocket连接关闭 - SessionId: {}, FromNo: {}, 原因: {}, 剩余在线数: {}",
                        session.getId(), fromNo, closeReason.getReasonPhrase(), getOnlineCount());
            }
        } catch (Exception e) {
            LOG.error("WebSocket连接关闭处理失败 - SessionId: {}, FromNo: {}", 
                    session != null ? session.getId() : "unknown", fromNo, e);
        }
    }

    /**
     * 1、接受chat内容
     * 2、获取chat用户信息
     * 3、处理聊听内容
     * 4、返回结束标识符
     */
    @OnMessage
    public void onMessage(@PathParam("toNo") String toNo, String message, Session session) {
        LOG.info("收到消息 - FromNo: {}, ToNo: {}, Size: {}", fromNo, toNo, message.length());
        
        try {
            // 1. 消息验证
            if (!validateMessage(message)) return;
            
            // 2. 解析消息
            WebSocketMessage wsMessage = parseMessage(message);
            if (wsMessage == null) return;
            
            // 3. 构建聊天请求并处理
            AppChatDTO chatRequest = buildChatRequest(wsMessage, toNo, session);
            processMessageAsync(chatRequest);
            
        } catch (Exception e) {
            LOG.error("消息处理失败 - FromNo: {}, ToNo: {}", fromNo, toNo, e);
//            sendMessage("{\"type\":\"error\",\"message\":\"处理失败\"}");
        }
    }

    /**
     * WebSocket错误处理
     */
    @OnError
    public void onError(Session session, Throwable error) {
        LOG.error("WebSocket连接发生错误 - SessionId: {}, FromNo: {}",
                session != null ? session.getId() : "unknown", fromNo, error);
        // 尝试关闭有问题的连接
        if (session != null && session.isOpen()) {
            try {
                session.close(new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, "服务器内部错误"));
            } catch (IOException e) {
                LOG.error("关闭错误连接失败", e);
            }
        }
    }

    /**
     * 异步处理消息 - 简化日志版本
     */
    private void processMessageAsync(AppChatDTO chatRequest) {
        try {
            WEB_SOCKET_EXECUTOR.submit(() -> {
                try {
                    WsServerEndpoint targetEndpoint = findWebSocketEndpoint(chatRequest.getFromNo(), chatRequest.getToNo());
                    if (targetEndpoint == null) {
                        LOG.error("未找到WebSocket连接 - FromNo: {}, ToNo: {}, 当前连接数: {}", 
                                chatRequest.getFromNo(), chatRequest.getToNo(), wsClientMap.size());
                        
                        // 尝试清理可能的僵尸连接
                        cleanupZombieConnectionsStatic();
                        
                        // 再次尝试查找
                        targetEndpoint = findWebSocketEndpoint(chatRequest.getFromNo(), chatRequest.getToNo());
                        if (targetEndpoint == null) {
                            LOG.error("清理后仍未找到连接 - FromNo: {}, ToNo: {}", 
                                    chatRequest.getFromNo(), chatRequest.getToNo());
                            // 发送错误消息给客户端
                            sendMessage("{\"type\":\"error\",\"message\":\"连接不存在，请重新连接\"}");
                            return;
                        } else {
                            LOG.info("清理后找到连接 - FromNo: {}, ToNo: {}", 
                                    chatRequest.getFromNo(), chatRequest.getToNo());
                        }
                    }
                    
                    AppAgentEngine appAgentChatBiz = applicationContext.getBean(AppAgentEngine.class);
                    LOG.info("开始智能体处理 - FromNo: {}, ToNo: {}", 
                            chatRequest.getFromNo(), chatRequest.getToNo());
                    
                    WebSocketAiChatAsyncBiz asyncBiz = new WebSocketAiChatAsyncBiz(chatRequest, targetEndpoint, appAgentChatBiz);
                    asyncBiz.run();

                } catch (Exception e) {
                    LOG.error("异步处理失败 - FromNo: {}, ToNo: {}", 
                             chatRequest.getFromNo(), chatRequest.getToNo(), e);
//                    sendMessage("{\"type\":\"error\",\"message\":\"处理失败\"}");
                }
            });
            
        } catch (RejectedExecutionException e) {
            LOG.error("线程池拒绝任务 - FromNo: {}, ToNo: {}", 
                     chatRequest.getFromNo(), chatRequest.getToNo(), e);
            sendMessage("{\"type\":\"error\",\"message\":\"服务器繁忙\"}");
        }
    }

    /**
     * 查找指定用户的WebSocket连接 - 增强状态验证
     */
    private WsServerEndpoint findWebSocketEndpoint(String fromNo, String toNo) {
        if (wsClientMap.isEmpty() || !StringUtils.hasText(fromNo)) {
            LOG.debug("查找连接失败 - wsClientMap为空或fromNo为空: {}", fromNo);
            return null;
        }
        
        WsServerEndpoint result = wsClientMap.stream()
                .filter(ws -> Objects.equals(ws.fromNo, fromNo))
                .filter(ws -> StringUtils.hasText(toNo) ? Objects.equals(ws.toNo, toNo) : true)
                .filter(ws -> ws.session != null && ws.session.isOpen()) // 添加连接状态验证
                .findFirst()
                .orElse(null);
                
        if (result == null) {
            LOG.warn("未找到有效的WebSocket连接 - FromNo: {}, ToNo: {}, 总连接数: {}", 
                    fromNo, toNo, wsClientMap.size());
            // 记录当前所有连接状态用于调试
            wsClientMap.forEach(ws -> 
                LOG.debug("现有连接 - FromNo: {}, ToNo: {}, SessionOpen: {}", 
                    ws.fromNo, ws.toNo, ws.session != null && ws.session.isOpen()));
        } else {
            LOG.debug("找到有效连接 - FromNo: {}, ToNo: {}, SessionId: {}", 
                    fromNo, toNo, result.session.getId());
        }
        
        return result;
    }

    /**
     * 提取用户信息
     */
    private UserInfo extractUserInfo(EndpointConfig config) {
        try {
            String tenantId = (String) config.getUserProperties().get("tenantId");
            String bizRoleType = (String) config.getUserProperties().get("bizRoleType");
            String bizRoleTypeNo = (String) config.getUserProperties().get("bizRoleTypeNo");
            String username = (String) config.getUserProperties().get("username");
            Boolean authenticated = (Boolean) config.getUserProperties().get("authenticated");
            if (!StringUtils.hasText(bizRoleTypeNo)) {
                LOG.warn("用户角色类型编号为空");
                return null;
            }
            return new UserInfo(tenantId, bizRoleType, bizRoleTypeNo, username,
                    authenticated != null ? authenticated : false);
        } catch (Exception e) {
            LOG.error("提取用户信息失败", e);
            return null;
        }
    }

    /**
     * 设置登录上下文
     */
    private void setupLoginContext(UserInfo userInfo) {
        LoginInfoContextHelper.set("tenantId", userInfo.getTenantId());
        LoginInfoContextHelper.set("bizRoleType", userInfo.getBizRoleType());
        LoginInfoContextHelper.set("bizRoleTypeNo", userInfo.getBizRoleTypeNo());
        LoginInfoContextHelper.set("username", userInfo.getUsername());
    }

    /**
     * 处理重复登录 - 修复竞态条件
     */
    private void handleDuplicateLogin(String bizRoleTypeNo) {
        synchronized (connectionLock) {
            Session existingSession = sessionPool.get(bizRoleTypeNo);
            if (existingSession != null && existingSession.isOpen()) {
                try {
                    LOG.info("检测到重复登录，关闭旧连接 - 用户: {}", bizRoleTypeNo);
                    // 先关闭连接，再移除，避免竞态条件
                    existingSession.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "重复登录"));
                    sessionPool.remove(bizRoleTypeNo);
                    
                    // 同时从wsClientMap中移除对应的连接
                    wsClientMap.removeIf(ws -> 
                        Objects.equals(ws.fromNo, bizRoleTypeNo) && 
                        ws.session != null && 
                        ws.session.equals(existingSession));
                    
                    LOG.info("重复登录处理完成 - 用户: {}", bizRoleTypeNo);
                } catch (IOException e) {
                    LOG.error("关闭重复登录连接失败 - 用户: {}", bizRoleTypeNo, e);
                    // 即使关闭失败也要移除，避免僵尸连接
                    sessionPool.remove(bizRoleTypeNo);
                    wsClientMap.removeIf(ws -> 
                        Objects.equals(ws.fromNo, bizRoleTypeNo) && 
                        ws.session != null && 
                        ws.session.equals(existingSession));
                }
            }
        }
    }

    /**
     * 注册新连接 - 添加同步保护
     */
    private void registerConnection(String bizRoleTypeNo, Session session) {
        synchronized (connectionLock) {
            sessionPool.put(bizRoleTypeNo, session);
            wsClientMap.add(this);
            onlineCount.incrementAndGet();
            LOG.debug("连接注册完成 - 用户: {}, SessionId: {}", bizRoleTypeNo, session.getId());
        }
    }

    /**
     * 安全关闭连接
     */
    private void closeWithReason(Session session, String reason) {
        try {
            if (session != null && session.isOpen()) {
                session.close(new CloseReason(CloseReason.CloseCodes.VIOLATED_POLICY, reason));
            }
        } catch (IOException e) {
            LOG.error("关闭WebSocket连接失败: {}", reason, e);
        }
    }

    /**
     * 发送消息到客户端
     */
    public void sendMessage(String message) {
        if (session == null || !session.isOpen()) {
            LOG.warn("Session已关闭，无法发送消息 - FromNo: {}", fromNo);
            return;
        }
        try {
            synchronized (session) {
                session.getBasicRemote().sendText(message);
            }
        } catch (IOException e) {
            LOG.error("发送消息失败 - FromNo: {}, Message: {}", fromNo, message, e);
        }
    }

    // 消息发送队列 - 线程安全
    private final Queue<String> messageQueue = new ConcurrentLinkedQueue<>();
    private volatile boolean isSending = false;
    private final Object sendLock = new Object();
    
    public void sendMessageAsync(String message) {
        if (message == null) {
            return;
        }
        
        // 检查连接是否已关闭
        if (closed) {
            LOG.warn("连接已关闭，丢弃消息 - FromNo: {}", fromNo);
            return;
        }
        
        // 检查连接状态
        if (!isConnectionValid()) {
            LOG.warn("WebSocket连接无效，丢弃消息 - FromNo: {}", fromNo);
            return;
        }
        
        // 检查队列大小，防止内存泄漏
        if (messageQueue.size() >= MAX_QUEUE_SIZE) {
            // 移除最旧的消息，为新消息腾出空间
            String oldMessage = messageQueue.poll();
            LOG.warn("消息队列已满({})，移除最旧消息 - FromNo: {}, 被移除消息长度: {}", 
                    MAX_QUEUE_SIZE, fromNo, oldMessage != null ? oldMessage.length() : 0);
        }
        
        // 添加到队列
        messageQueue.offer(message);
        LOG.debug("消息入队 - FromNo: {}, 当前队列大小: {}/{}", fromNo, messageQueue.size(), MAX_QUEUE_SIZE);
        
        // 尝试发送队列中的消息
        processMessageQueue();
    }
    
    /**
     * 检查连接是否有效
     */
    private boolean isConnectionValid() {
        return session != null && session.isOpen() && 
               session.getAsyncRemote() != null && 
               session.getBasicRemote() != null;
    }
    
    /**
     * 处理消息队列 - 确保消息按顺序发送
     */
    private void processMessageQueue() {
        // 如果连接已关闭，不再处理队列
        if (closed) {
            LOG.debug("连接已关闭，停止处理消息队列 - FromNo: {}", fromNo);
            return;
        }
        
        synchronized (sendLock) {
            if (isSending || messageQueue.isEmpty()) {
                return;
            }
            isSending = true;
        }
        
        // 异步处理消息队列
        WEB_SOCKET_EXECUTOR.execute(() -> {
            try {
                while (!messageQueue.isEmpty() && !closed) {
                    String message = messageQueue.poll();
                    if (message != null && session != null && session.isOpen()) {
                        sendMessageSafely(message);
                    } else if (closed) {
                        // 连接已关闭，丢弃剩余消息
                        LOG.debug("连接已关闭，丢弃剩余消息 - FromNo: {}, 剩余消息数: {}", 
                                fromNo, messageQueue.size());
                        break;
                    }
                }
            } finally {
                synchronized (sendLock) {
                    isSending = false;
                    // 如果队列中还有消息且连接未关闭，继续处理
                    if (!messageQueue.isEmpty() && !closed) {
                        processMessageQueue();
                    }
                }
            }
        });
    }
    
    /**
     * 安全发送消息 - 添加重试机制
     */
    private void sendMessageSafely(String message) {
        int maxRetries = 3;
        int retryCount = 0;
        
        while (retryCount < maxRetries) {
            try {
                if (!session.isOpen()) {
                    LOG.warn("WebSocket连接已关闭，无法发送消息");
                    return;
                }
                
                // 检查连接状态
                if (session.getAsyncRemote() == null) {
                    LOG.warn("WebSocket异步远程端点不可用");
                    return;
                }
                
                // 使用同步发送避免状态冲突
                session.getBasicRemote().sendText(message);
                LOG.debug("消息发送成功 - FromNo: {}, Length: {}", fromNo, message.length());
                return;
                
            } catch (Exception e) {
                retryCount++;
                LOG.warn("消息发送失败，重试 {}/{} - FromNo: {}, Error: {}", 
                    retryCount, maxRetries, fromNo, e.getMessage());
                
                if (retryCount >= maxRetries) {
                    LOG.error("消息发送最终失败 - FromNo: {}, Message: {}", fromNo, message, e);
                } else {
                    // 等待一段时间后重试
                    try {
                        Thread.sleep(100 * retryCount);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }
    }

    // 静态方法 - 获取统计信息
    public static int getOnlineCount() {
        return onlineCount.get();
    }

    public static CopyOnWriteArraySet<WsServerEndpoint> getWsClientMap() {
        return new CopyOnWriteArraySet<>(wsClientMap); // 返回副本，避免外部修改
    }

    public static ConcurrentHashMap<String, Session> getSessionPool() {
        return new ConcurrentHashMap<>(sessionPool); // 返回副本，避免外部修改
    }

    /**
     * 优雅关闭 - 应用停止时调用
     */
    @PreDestroy
    public void shutdown() {
        LOG.info("开始优雅关闭WebSocket服务...");
        
        // 1. 关闭所有WebSocket连接并清理消息队列
        int totalQueuedMessages = 0;
        for (WsServerEndpoint endpoint : wsClientMap) {
            try {
                // 标记连接已关闭
                endpoint.closed = true;
                
                // 清理消息队列，防止内存泄漏
                int queueSize = endpoint.messageQueue.size();
                totalQueuedMessages += queueSize;
                endpoint.messageQueue.clear();
                
                if (endpoint.session != null && endpoint.session.isOpen()) {
                    endpoint.session.close(new CloseReason(
                        CloseReason.CloseCodes.GOING_AWAY, "服务器正在关闭"));
                }
            } catch (IOException e) {
                LOG.error("关闭WebSocket连接失败 - FromNo: {}", endpoint.fromNo, e);
            }
        }
        
        if (totalQueuedMessages > 0) {
            LOG.info("清理了 {} 个未发送的消息", totalQueuedMessages);
        }
        
        // 2. 关闭线程池
        WEB_SOCKET_EXECUTOR.shutdown();
        try {
            // 等待60秒让任务完成
            if (!WEB_SOCKET_EXECUTOR.awaitTermination(60, TimeUnit.SECONDS)) {
                LOG.warn("线程池未能在60秒内关闭，强制关闭");
                WEB_SOCKET_EXECUTOR.shutdownNow();
                // 再等待30秒
                if (!WEB_SOCKET_EXECUTOR.awaitTermination(30, TimeUnit.SECONDS)) {
                    LOG.error("线程池强制关闭失败");
                }
            }
        } catch (InterruptedException e) {
            LOG.warn("等待线程池关闭被中断");
            WEB_SOCKET_EXECUTOR.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        LOG.info("WebSocket服务关闭完成");
    }

    /**
     * 获取线程池状态信息 - 用于监控
     */
    public static String getThreadPoolStatus() {
        return String.format("ThreadPool[Active: %d, Pool: %d, Queue: %d, Completed: %d]",
            WEB_SOCKET_EXECUTOR.getActiveCount(),
            WEB_SOCKET_EXECUTOR.getPoolSize(),
            WEB_SOCKET_EXECUTOR.getQueue().size(),
            WEB_SOCKET_EXECUTOR.getCompletedTaskCount());
    }

    /**
     * 清理僵尸连接 - 移除无效的WebSocket连接（静态方法）
     */
    private static void cleanupZombieConnectionsStatic() {
        try {
            synchronized (connectionLock) {
                int beforeCount = wsClientMap.size();
                
                // 清理wsClientMap中的僵尸连接
                wsClientMap.removeIf(ws -> {
                    boolean isZombie = ws.session == null || !ws.session.isOpen();
                    if (isZombie) {
                        LOG.debug("清理僵尸连接 - FromNo: {}, ToNo: {}", ws.fromNo, ws.toNo);
                    }
                    return isZombie;
                });
                
                // 清理sessionPool中的僵尸连接
                sessionPool.entrySet().removeIf(entry -> {
                    boolean isZombie = entry.getValue() == null || !entry.getValue().isOpen();
                    if (isZombie) {
                        LOG.debug("清理僵尸Session - 用户: {}", entry.getKey());
                    }
                    return isZombie;
                });
                
                int afterCount = wsClientMap.size();
                int cleanedCount = beforeCount - afterCount;
                
                if (cleanedCount > 0) {
                    LOG.info("清理僵尸连接完成 - 清理数量: {}, 剩余连接: {}", cleanedCount, afterCount);
                }
            }
        } catch (Exception e) {
            LOG.error("清理僵尸连接失败", e);
        }
    }

    /**
     * 清理僵尸连接 - 移除无效的WebSocket连接（实例方法）
     */
    private void cleanupZombieConnections() {
        cleanupZombieConnectionsStatic();
    }



    /**
     * 消息验证 - 简洁版本
     */
    private boolean validateMessage(String message) {
        if (!StringUtils.hasText(message)) {
            LOG.warn("空消息 - FromNo: {}", fromNo);
            return false;
        }
        
        if (message.length() > MAX_MESSAGE_SIZE) {
            LOG.warn("消息过大 - FromNo: {}, Size: {}", fromNo, message.length());
            sendMessage("{\"type\":\"error\",\"message\":\"消息过大\"}");
            return false;
        }
        
        long currentTime = System.currentTimeMillis();
        if (currentTime - lastMessageTime < MESSAGE_INTERVAL_MS) {
            LOG.warn("频率过高 - FromNo: {}", fromNo);
            return false;
        }
        lastMessageTime = currentTime;
        
        return true;
    }

    /**
     * 解析WebSocket消息 - 简洁版本
     */
    private WebSocketMessage parseMessage(String message) {
        try {
            JsonNode messageJson = OBJECT_MAPPER.readTree(message);
            String messageType = messageJson.path("type").asText("");
            
            // 忽略ping消息
            if ("ping".equals(messageType)) {
                return null;
            }
            
            // 只处理ai_chat消息
            if (!"ai_chat".equals(messageType)) {
                LOG.warn("不支持的消息类型: {} - FromNo: {}", messageType, fromNo);
                return null;
            }
            
            // 提取内容
            String content = extractContent(messageJson);
            if (!StringUtils.hasText(content)) {
                LOG.warn("消息内容为空 - FromNo: {}", fromNo);
                return null;
            }
            
            return new WebSocketMessage(messageType, content);
            
        } catch (Exception e) {
            LOG.error("消息解析失败 - FromNo: {}", fromNo, e);
            sendMessage("{\"type\":\"error\",\"message\":\"消息格式错误\"}");
            return null;
        }
    }

    /**
     * 提取消息内容 - 兼容多种格式
     */
    private String extractContent(JsonNode messageJson) {
        // 优先从payload.content获取
        if (messageJson.has("payload") && messageJson.get("payload").has("content")) {
            return messageJson.get("payload").get("content").asText();
        }
        // 兼容直接content字段
        if (messageJson.has("content")) {
            return messageJson.get("content").asText();
        }
        return null;
    }

    /**
     * 构建聊天请求 - 简洁版本
     */
    private AppChatDTO buildChatRequest(WebSocketMessage wsMessage, String toNo, Session session) {
        String tenantId = (String) session.getUserProperties().get("tenantId");
        String bizRoleType = (String) session.getUserProperties().get("bizRoleType");
        String bizRoleTypeNo = (String) session.getUserProperties().get("bizRoleTypeNo");
        String currentFromNo = Optional.ofNullable(LoginInfoContextHelper.getLoginUser())
                .map(user -> user.getBizRoleTypeNo())
                .orElse(fromNo);
        return AppChatDTO.builder()
                .tenantId(tenantId)
                .bizRoleType(bizRoleType)
                .bizRoleTypeNo(bizRoleTypeNo)
                .conversationNo(String.format("%s_%s", BsinSnowflake.getId(), session.getId()))
                .chatType(chatType)
                .question(wsMessage.getContent())
                .fromNo(currentFromNo)
                .toNo(toNo)
                .build();
    }

    // 内部消息类
    private static class WebSocketMessage {
        private final String type;
        private final String content;

        public WebSocketMessage(String type, String content) {
            this.type = type;
            this.content = content;
        }

        public String getType() { return type; }
        public String getContent() { return content; }
    }

    // 内部用户信息类
    private static class UserInfo {
        private final String tenantId;
        private final String bizRoleType;
        private final String bizRoleTypeNo;
        private final String username;
        private final boolean authenticated;
        
        public UserInfo(String tenantId, String bizRoleType, String bizRoleTypeNo, String username, boolean authenticated) {
            this.tenantId = tenantId;
            this.bizRoleType = bizRoleType;
            this.bizRoleTypeNo = bizRoleTypeNo;
            this.username = username;
            this.authenticated = authenticated;
        }

        public String getTenantId() { return tenantId; }
        public String getBizRoleType() { return bizRoleType; }
        public String getBizRoleTypeNo() { return bizRoleTypeNo; }
        public String getUsername() { return username; }
        public boolean isAuthenticated() { return authenticated; }
    }

}